package com.atguigu.chapter08;

import com.atguigu.Bean.LoginEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;

/**
 * @ClassName: Flink06_Login_Project
 * @Description:  恶意登录监控
 * @Author: kele
 * @Date: 2021/4/10 10:50
 *
 * 需求：如果同一用户（可以是不同IP）在2秒之内连续两次登录失败，就认为存在恶意登录的风险，
 *
 **/
public class Flink06_Login_Project {


    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);


        env
//                .socketTextStream("hadoop162",8888)
                .readTextFile("in/LoginLog.csv")
                .map(line -> {
                    String[] data = line.split(",");
                    return new LoginEvent(Long.valueOf(data[0]),
                            data[1],
                            data[2],
                            Long.parseLong(data[3]) * 1000L);
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.
                                <LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                        .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                            @Override
                            public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                                return element.getEventTime();
                            }
                        })
                )
                .keyBy(LoginEvent::getUserId)
               .process(new KeyedProcessFunction<Long, LoginEvent, String>() {

                   private ListState<Long> failLoginState;

                   @Override
                   public void open(Configuration parameters) throws Exception {
                       failLoginState = getRuntimeContext()
                               .getListState(new ListStateDescriptor<Long>("FailLoginState", Long.class));
                   }

                   @Override
                   public void processElement(LoginEvent value, Context ctx, Collector<String> out) throws Exception {

                       //定义一个集合，将数据按照时间进行排序，
                       ArrayList<Long> loginEventList = new ArrayList<>();

//                       loginEventList.sort((o1, o2) -> o1.compareTo(o2));
                       loginEventList.sort(Long::compareTo);

                       if("fail".equalsIgnoreCase(value.getEventType())){

                           failLoginState.add(value.getEventTime());

                           for (Long aLong : failLoginState.get()) {
                               loginEventList.add(aLong);
                           }

                           if( loginEventList.size() == 2 ) {

                               //如果集合中的数据差小于
                               if (loginEventList.get(1) - loginEventList.get(0) <= 2000) {
                                   out.collect(ctx.getCurrentKey() + "两秒内连续两次登录失败，报警......");
                               }

                               //集合中保存了两个，则删除一个,更新状态
                               loginEventList.remove(0);
                               failLoginState.update(loginEventList);
                           }

                       }
                       //如果成功了，则将集合状态中的元素清空
                       else{
                           failLoginState.clear();
                       }
                   }
               })
                .print();



        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
